package com.fluidops.datacatalog.reltables.data;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;

import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.MultiThreadedHttpConnectionManager;
import org.apache.log4j.Logger;
import org.openrdf.model.Literal;
import org.openrdf.model.Statement;
import org.openrdf.model.URI;
import org.openrdf.model.Value;
import org.openrdf.model.ValueFactory;
import org.openrdf.model.impl.ValueFactoryImpl;
import org.openrdf.model.vocabulary.RDF;
import org.openrdf.model.vocabulary.RDFS;
import org.openrdf.query.BindingSet;
import org.openrdf.query.QueryLanguage;
import org.openrdf.query.TupleQuery;
import org.openrdf.query.TupleQueryResult;
import org.openrdf.repository.Repository;
import org.openrdf.repository.RepositoryConnection;
import org.openrdf.repository.sail.SailRepository;
import org.openrdf.rio.RDFFormat;
import org.openrdf.sail.memory.MemoryStore;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.xml.sax.SAXException;

import com.fluidops.datacatalog.reltables.Starter;
import com.fluidops.datacatalog.reltables.data.Vocabulary.DCTERMS;
import com.fluidops.datacatalog.reltables.data.Vocabulary.DataCube;
import com.fluidops.datacatalog.reltables.data.Vocabulary.SKOS;


public class EurostatLabelProvider {

	public static final String metaXMLNamespace = "nt:";

	public static final String eurostatRDFNamespace = "http://eurostat.linked-statistics.org/";

	public static final String metaURLXML = "http://epp.eurostat.ec.europa.eu/NavTree_prod/everybody/BulkDownloadListing?sort=1&file=table_of_contents.xml";

	private static Logger logger = Logger.getLogger(EurostatLabelProvider.class.getName());

	public EurostatLabelProvider(){

	}


	public List<Statement> getLabelStatements() throws Exception{
		Set<Statement> labels = new HashSet<Statement>();
		ValueFactory vf = ValueFactoryImpl.getInstance();
		List<EurostatMetaDescription> metaDesc = getMetaFromXML(metaURLXML);

		MultiThreadedHttpConnectionManager connectionManager = new MultiThreadedHttpConnectionManager();
		HttpClient client = new HttpClient(connectionManager);
		ExecutorService pool =  Executors.newFixedThreadPool(10);

		Repository repository = new SailRepository(new MemoryStore());
		repository.initialize();

		for(EurostatMetaDescription desc : metaDesc){
			// as an indicator if the data set is existing, some are just part of the meta file...
			if(desc.lastUpdate != null){
				URI datasetURI = vf.createURI(eurostatRDFNamespace, "data/" + desc.code);
				URI schemaURI = vf.createURI(eurostatRDFNamespace, "dsd/" + desc.code);

				labels.add(vf.createStatement(datasetURI, DCTERMS.TITLE, vf.createLiteral(desc.title)));
				if(desc.desc != null){

					labels.add(vf.createStatement(datasetURI, Vocabulary.DCTERMS.DESCRIPTION, vf.createLiteral(desc.desc)));
				}
				labels.add(vf.createStatement(datasetURI, RDF.TYPE , DataCube.DATASET));

				/*
				 *register download for DSD file will be used after the loop
				 */
				RepositoryConnection connection = repository.getConnection();
				pool.execute(new MetaReader(connection, client, schemaURI.stringValue() + ".ttl", schemaURI, RDFFormat.TURTLE));
			}
		}

		logger.debug("Meta information for data sets created.");

		pool.shutdown();
		pool.awaitTermination(4, TimeUnit.HOURS);
		logger.debug("DSD files downloaded.");

		RepositoryConnection connection = repository.getConnection();
		for(EurostatMetaDescription desc : metaDesc){

			if(desc.lastUpdate != null){
				URI schemaURI = vf.createURI(eurostatRDFNamespace, "dsd/" + desc.code);
				URI datasetURI = vf.createURI(eurostatRDFNamespace, "data/" + desc.code);
				/*
				 * SELECT ?dimension ?label ?dimType ?values ?valueLabel
				 * FROM <schemaURI>
				 * WHERE{
				 *      <schemaURI> qb:component _:b1 .
				 *      _:b1 ?dimType ?dimension .  ## eg: _:b1 qb:attribue property:obs_status
				 *      
				 *      OPTIONAL{
				 *         ?dimension qb:codeList ?dicDimension .
				 *         ?dicDimension rdfs:label ?label ;  ## eg:"ObservationStatus"
				 *                       skos:topConcept ?values .
				 *         ?values skos:prefLabel ?valueLabel .
				 *      }
				 * }
				 */
				StringBuilder query = new StringBuilder(" SELECT ?dimension ?label ?values ?valueLabel ");
				query.append(" FROM <").append(schemaURI.stringValue()).append("> ")
				.append(" WHERE { <").append(schemaURI.stringValue()).append("> <").append(DataCube.COMPONENT.stringValue()).append("> _:b1 .")
				.append(" _:b1 ").append(" ?dimType ?dimension .")
				.append(" OPTIONAL { ")
				.append(" ?dimension <").append(DataCube.CODE_LIST.stringValue()).append("> ?dicDimension .")
				.append(" ?dicDimension <").append(RDFS.LABEL.stringValue()).append("> ?label ; <")
				.append(SKOS.HAS_TOP_CONCEPT.stringValue()).append("> ?values .")
				.append(" ?values <").append(SKOS.PREFLABEL.stringValue()).append("> ?valueLabel . } }");


				TupleQuery tquery = connection.prepareTupleQuery(QueryLanguage.SPARQL, query.toString());
				TupleQueryResult result = tquery.evaluate();
				// to avoid duplicates in the res list (would be filtered afterwards but saves memory)
				Set<Value> duplicateFilter = new HashSet<Value>();
				try{
					BindingSet bs;
					URI dim;
					Literal label;
					while(result.hasNext()){
						bs = result.next();
						dim = (URI) bs.getValue("dimension");

						// example:  <dsd:aact_ali02> qb:dimension property:unit .
						if(bs.hasBinding("label") && !duplicateFilter.contains(bs.getValue("label"))){
							label = (Literal) bs.getValue("label");
							if("en".equals(label.getLanguage())){
								labels.add(vf.createStatement(datasetURI, dim, label));
								duplicateFilter.add(label);
							}
						}
						if(!duplicateFilter.contains(dim)){
							labels.add(vf.createStatement(datasetURI, DataCube.DIMENSION, dim));
							labels.add(vf.createStatement(dim, RDF.TYPE, DataCube.DIMENSION_PROPERTY));
							duplicateFilter.add(dim);
						}
						if(bs.hasBinding("values") && bs.hasBinding("valueLabel")){
							if(!duplicateFilter.contains(bs.getValue("valueLabel"))){
								label = (Literal) bs.getValue("valueLabel");
								if("en".equals(label.getLanguage())){
									labels.add(vf.createStatement(datasetURI, dim, bs.getValue("values")));
									labels.add(vf.createStatement((URI) bs.getValue("values"), SKOS.PREFLABEL, bs.getValue("valueLabel")));
									duplicateFilter.add(bs.getValue("valueLabel"));
								}
							}

						}
					}
				}finally{
					if(result != null){
						result.close();
					}
				}
			}
		}
		connection.close();
		labels.add(vf.createStatement(vf.createURI(Starter.catalogSubject), vf.createURI(Starter.catalogProperty), vf.createURI(eurostatRDFNamespace)));

		return new ArrayList<Statement>(labels);
	}


	private List<EurostatMetaDescription> getMetaFromXML(String metaDataLocation){
		List<EurostatMetaDescription> metaList = new LinkedList<EurostatMetaDescription>();
		// retrieve meta data document
		String content = null;
		// check if the meta file is retrieved from the web or from data dump
		URL metaURL = null;
		try {
			metaURL = new URL(metaDataLocation);
		} catch (MalformedURLException e1) {
			logger.warn("Malformed MetaData(XML) URL for Eurostat.", e1);
			throw new RuntimeException(e1.getMessage()); 
		}
		if(metaURL != null){
			content  = downloadAsString(metaURL);
		}
		if(content !=null){
			// parse and fill the list
			parseAndRetrieveMetaXML(metaList, content);
		}
		return metaList;
	}


	private void parseAndRetrieveMetaXML(List<EurostatMetaDescription> metaList, String content){
		DocumentBuilder xmlBuilder = null;
		try {
			xmlBuilder = DocumentBuilderFactory.newInstance().newDocumentBuilder();
			// ensure the character format is utf-8 otherwise the xml-parsing will break
			ByteArrayInputStream encXML = new  ByteArrayInputStream(content.getBytes("UTF8"));
			Document doc = xmlBuilder.parse(encXML);
			/*
			 * Example snippet:
			 * 
			 * <nt:leaf type="dataset">
			 *   <nt:title language="en">Road, rail and navigable inland waterways networks by NUTS 2 regions</nt:title>
			 *   <nt:code>tran_r_net</nt:code>
			 *   <nt:lastUpdate>14.05.2012</nt:lastUpdate>
			 * </nt:leaf>
			 * 
			 */
			NodeList nodeList = doc.getElementsByTagName(metaXMLNamespace + "leaf");
			//NodeList nodeList = doc.getElementsByTagNameNS(metaXMLNamespace, "leaf");
			EurostatMetaDescription desc;
			for(int i = 0; i < nodeList.getLength(); i++){
				Node node = nodeList.item(i);
				if(node != null && node.getAttributes() != null){// && node.getAttributes().item(0).getTextContent().equals("dataset")){
					Element elem = (Element) node;
					desc = new EurostatMetaDescription();
					if(elem != null){
						desc.title= getElementValue("title", elem);
						desc.code = getElementValue("code", elem);
						desc.lastUpdate = getElementValue("lastUpdate", elem);
						desc.desc = getElementValue("shortDescription", elem);
						metaList.add(desc);
					}
				}
			}

		} catch (ParserConfigurationException e) {
			logger.warn("Could not instantiate XML-DocumentBuilder", e);
			throw new RuntimeException(e.getMessage());
		} catch (SAXException e) {
			logger.warn("Any XML-Exception occured while parsing Eurostat meta file.", e);
			throw new RuntimeException(e.getMessage());
		} catch (IOException e) {
			logger.warn("An IO Exception occured while parsing Eurostat meta file.", e);
			throw new RuntimeException(e.getMessage());
		}
	}


	private String getElementValue(String tag, Element elem){
		String val = null;
		NodeList nl = elem.getElementsByTagName(metaXMLNamespace + tag).getLength() > 0 ? elem.getElementsByTagName("nt:" + tag).item(0).getChildNodes() : null;
		if(nl != null){
			Node node = (Node) nl.item(0);
			val = node != null ? node.getTextContent() : null;
		}
		return val;

	}

	private String downloadAsString(URL url) {
		HttpURLConnection httpConnection = null;
		String content = null;
		try {
			httpConnection = (HttpURLConnection) url.openConnection();
			httpConnection.setRequestMethod("GET");
			if(httpConnection.getResponseCode() == HttpURLConnection.HTTP_OK){
				//download the meta file as a string
				content = readURL(httpConnection.getInputStream());
				httpConnection.disconnect();
			}
		} catch (IOException e1) {
			logger.warn("Could not establish link to " + url.toString(), e1);
			throw new RuntimeException(e1.getMessage()); 
		}
		return content;
	}

	private static String readURL( InputStream in) throws IOException
	{
		ByteArrayOutputStream out = null;
		try {
			byte[] buffer = new byte[8192];
			out = new ByteArrayOutputStream();

			while (true)
			{
				int len = in.read( buffer );
				if ( len == -1 )
					break;
				out.write( buffer, 0, len );
			}
		}
		finally {
			if(out != null){
				out.close();
			}
		}
		return out.toString();
	}



	private static class EurostatMetaDescription {

		public String title;
		public String code;
		public String desc;

		public String lastUpdate;

	}
}
